package com.ys.web.ws.send.service.impl;

import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.core.util.IdUtil;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.ys.base.BaseConstants;
import com.ys.base.EnumType;
import com.ys.utils.base.http.McodeHandler;
import com.ys.utils.base.j2se.*;
import com.ys.utils.help.GroupUtils;
import com.ys.utils.help.TipUtils;
import com.ys.utils.help.UserUtils;
import com.ys.utils.push.JPushUtils;
import com.ys.utils.redis.Redis;
import com.ys.utils.redis.key.GroupKey;
import com.ys.utils.redis.key.UserKey;
import com.ys.utils.spring.config.SystemError;
import com.ys.web.user.friends.manager.FriendsManager;
import com.ys.web.user.friends.mapper.GroupInfoMapper;
import com.ys.web.user.friends.mapper.UserApplyMapper;
import com.ys.web.user.friends.mapper.UserFriendsMapper;
import com.ys.web.user.friends.mapper.UserGroupMapper;
import com.ys.web.user.friends.po.BaseGroupInfoPO;
import com.ys.web.user.friends.po.BaseUserApplyPO;
import com.ys.web.user.friends.po.BaseUserFriendsPO;
import com.ys.web.user.login.LoginInfo;
import com.ys.web.user.manager.UserLogoutManager;
import com.ys.web.verticel.WebSocketVerticle;
import com.ys.web.ws.bean.MessageRequestBean;
import com.ys.web.ws.bean.MessageResponseBean;
import com.ys.web.ws.manager.ChatObjectManager;
import com.ys.web.ws.send.mapper.GChatMapper;
import com.ys.web.ws.send.mapper.GChatReadMapper;
import com.ys.web.ws.send.mapper.SChatMapper;
import com.ys.web.ws.send.po.GChatPO;
import com.ys.web.ws.send.po.GChatReadPO;
import com.ys.web.ws.send.po.SChatPO;
import com.ys.web.ws.send.service.MessageService;
import com.ys.web.ws.send.vo.QueryHistoryVo;
import io.vertx.core.http.ServerWebSocket;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.concurrent.Future;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * @ 名称:
 * @ 创建人: dhf
 * @ 参数： null
 * @ 创建时间: 2022/5/11
 * @ 修改人和其它信息
 * @ 返回值:
 */
@Slf4j
@Service
public class MessageServiceImpl implements MessageService {

    @Autowired
    private SChatMapper sChatMapper;

    @Autowired
    private UserFriendsMapper userFriendsMapper;

    @Autowired
    private UserApplyMapper userApplyMapper;

    @Autowired
    private GChatMapper gChatMapper;

    @Autowired
    private GChatReadMapper gChatReadMapper;

    @Autowired
    private UserGroupMapper userGroupMapper;

    @Autowired
    private GroupInfoMapper groupInfoMapper;


    private void assembly(MessageRequestBean messageRequestBean) {
        if (messageRequestBean.getCode().equals(EnumType.MessageCode.MESSAGE_SEND_USER_CARD.getCode())) {
            long id = McodeHandler.from32Radix(messageRequestBean.getData());
            LoginInfo cacheUser = UserUtils.getCacheUser(id);
            if (StringHandler.isEmpty(cacheUser.getAvatarUrl())) {
                cacheUser.setAvatarUrl("https://oss.shuzhuang.xyz/sys/register/pic/headInit.jpg");
            }
            messageRequestBean.setData(JsonUtil.toJson(MapHandler.initMap("nickName", cacheUser.getNickName(),
                    "alias", messageRequestBean.getData(), "avatarUrl", cacheUser.getAvatarUrl())));
        }
    }


    @Override
    public void sendMessageUser(MessageRequestBean messageRequestBean,long uid) {
        String userId = String.valueOf(McodeHandler.from32Radix(messageRequestBean.getFrom()));
        if (!messageRequestBean.getFrom().equals(UserUtils.ASSISTANT.getUserId()) && !messageRequestBean.getFrom().equals(UserUtils.CLIENT.getUserId()) && !messageRequestBean.getFrom().equals(UserUtils.SYS_NOTICE.getUserId())) {
            UserUtils.validateUserAccountStatus(userId);
        }
        String friendId = String.valueOf(McodeHandler.from32Radix(messageRequestBean.getTo()));
        if (!messageRequestBean.getFrom().equals(UserUtils.CLIENT.getUserId()) && !messageRequestBean.getTo().equals(UserUtils.CLIENT.getUserId())) {
            //判断好友是否注销
            int account_status = UserLogoutManager.account_status(StringHandler.getLong(friendId));
            if(account_status>0){
                if(account_status==1)
                    SystemError.wrapBs("500", "好友已注销");
                if(account_status==2)
                    SystemError.wrapBs("500", "好友已禁用");
            }
            BaseUserFriendsPO q1 = new BaseUserFriendsPO();
            q1.setUserId(userId);
            q1.setFriendId(friendId);
            q1.setStatus(true);
            BaseUserFriendsPO o1 = userFriendsMapper.selectOne(q1);
            if (o1 == null) {
                SystemError.wrapBs("500", "开启了朋友验证，你还不是他(她)朋友，请先发送朋友验证请求，对方验证通过后，才能聊天。");
            }

            if (FriendsManager.isBlackFriend(friendId, StringHandler.getLong(userId)) == 1) {
                SystemError.wrapBs("500", "你发出的消息已被拒收");
            }
        }
        this.assembly(messageRequestBean);
        SChatPO sChatPo = UserUtils.getSChatPO(messageRequestBean);
        sChatPo.setCreateTime(LocalDateTime.now().format(DateTimeFormatter.ofPattern(BaseConstants.BASE_FORMAT)));

        //处理用户消息
        ServerWebSocket serverWebSocket = ChatObjectManager.getConnectionMap().get(messageRequestBean.getTo());
        boolean flag = false;
        //组装消息对象
        MessageResponseBean responseBean = this.getSingleChat(messageRequestBean.getMessageId(),
                messageRequestBean.getFrom(), BaseConstants.SINGLE_PREFIX + messageRequestBean.getFrom(), messageRequestBean.getAvatarUrl(),
                messageRequestBean.getUserName(), messageRequestBean.getData(), messageRequestBean.getCreateTime(), messageRequestBean.getCode(),
                messageRequestBean.getWidth(), messageRequestBean.getHeight(), messageRequestBean.getSoundSeconds(),messageRequestBean.getLevel(),
                messageRequestBean.getPayChannel()
        );
        if (serverWebSocket != null) {
            String jsonStr = JsonUtil.toJson(new ArrayList<>(Collections.singletonList(responseBean)));
            try {
//                if (currentTime - ChatObjectManager.getConnectionMapTime().get(messageRequestBean.getTo()) <= 1500) {
//                if (!ChatObjectManager.getConnectionUnReadPage().contains(StringHandler.getLong(friendId))) {
                flag = WebSocketVerticle.sendMessage(jsonStr, serverWebSocket);
//                }
//                } else {
//                    serverWebSocket.close();
//                    ChatObjectManager.removeChatMapByKey(messageRequestBean.getTo());
//                }
            } catch (Exception e) {
                Logger.info("发送私聊消息出现异常：消息编号【" + messageRequestBean.getCode() + "],接收人：" + friendId);
            }
            sChatPo.setIsSuccess(false);
        }

        String userUnReadMessageEntityKey = UserKey.userUnReadMessageEntityKey(friendId);
        Redis.hash.hSet(userUnReadMessageEntityKey, sChatPo.getMessageId(), JsonUtil.toJson(responseBean));

        boolean finalFlag = flag;
        ThreadUtil.execAsync(() -> {
            //如果 socket 发送失败
            if (!finalFlag) {
                Set<String> members = new HashSet<>();
                String friendJson = (String) Redis.hash.hGet(UserKey.getUserFriendsListKey(friendId), userId);
                BaseUserFriendsPO baseUserFriendsPO = JsonUtil.jsonToBean(friendJson, BaseUserFriendsPO.class);
                if (messageRequestBean.getTo().equals(UserUtils.CLIENT.getUserId())) {
                    members.add(friendId);
                } else {
                    //判断好友是否开启了消息免打扰
                    if (StringHandler.isNotEmpty(baseUserFriendsPO) && StringHandler.isNotEmpty(baseUserFriendsPO.getId())) {
                        //未开起
                        if (StringHandler.isEmpty(baseUserFriendsPO.getDisturb()) || !baseUserFriendsPO.getDisturb()) {
                            //发送推送
                            //发送消息
                            members.add(friendId);
                        }
                    }
                }
                if (!members.isEmpty()) {
                    JPushUtils.sendNotification(members, messageRequestBean);
                }else{
                    //客服发给用户
                    if(uid==0 && McodeHandler.to32Radix(uid).equals(messageRequestBean.getFrom())){
                        members.add(friendId);
                        JPushUtils.sendNotification(members, messageRequestBean);
                    }
                    ////用户发给客服
                    if(McodeHandler.from32Radix(messageRequestBean.getTo())==0){
                        members.add(friendId);
                        JPushUtils.sendNotification(members, messageRequestBean);
                    }
                }
            }
        });

    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void sendMessageGroup(MessageRequestBean messageRequestBean) {
        long currentTime = System.currentTimeMillis();
        //判断用户是否还在群中
        String fromId = String.valueOf(McodeHandler.from32Radix(messageRequestBean.getFrom()));
        if (!messageRequestBean.getFrom().equals(UserUtils.ASSISTANT.getUserId()) && !messageRequestBean.getFrom().equals(UserUtils.CLIENT.getUserId()) && !messageRequestBean.getFrom().equals(UserUtils.SYS_NOTICE.getUserId())) {
            UserUtils.validateUserAccountStatus(fromId);
        }
        //判断群组是否存在
        //获取缓存key
        String key = GroupKey.groupMembersKey(messageRequestBean.getGroupId());
        Set<String> members = Redis.zSet.range(key, 0, -1);
        if (members == null || members.isEmpty()) {
            SystemError.wrapBs(EnumType.ResponseCode.Fail.getCode(), "当前群已解散");
        } else {
            if (!messageRequestBean.getFrom().equals(UserUtils.ASSISTANT.getUserId()) && !members.contains(fromId)) {
                SystemError.wrapBs(EnumType.ResponseCode.Fail.getCode(), "您已不再该群，无法发送消息");
            }
        }

        BaseGroupInfoPO baseGroupInfoPO = GroupUtils.getCacheGroupInfo(messageRequestBean.getGroupId());
        if (!StringHandler.isEmpty(baseGroupInfoPO.getIsFreeze()) && baseGroupInfoPO.getIsFreeze()) {
            SystemError.wrapBs(EnumType.ResponseCode.Fail.getCode(), "该群聊已被封禁");
        }

        this.assembly(messageRequestBean);
        GChatPO gChatPO = UserUtils.getGChatPO(messageRequestBean);
//            gChatMapper.insertSelective(gChatPO);

        //查找群成员用户  是否开启免打扰
        Future<Map<String, Boolean>> userSetMap = ThreadUtil.execAsync(() -> {
            List<Map<String, String>> collect = userGroupMapper.findGroupUserSetDisturb(messageRequestBean.getGroupId());
            Map<String, Boolean> map = collect.stream().collect(Collectors.toMap(x -> String.valueOf(x.get("userId")), x -> Boolean.parseBoolean(x.get("disturb"))));
            return map;
        });

        //获取所有人的终端
//        Future<Map<String, String>> terminalMap = ThreadUtil.execAsync(() -> {
//            Map<Object, Object> all = Redis.hash.hgetall(UserKey.terminalHash());
//            Map<String, String> collect = all.entrySet().parallelStream().collect(Collectors.toMap(x -> String.valueOf(x.getKey()), x -> String.valueOf(x.getValue())));
//            return collect;
//        });


        //排除发信息的那个人
        List<String> ids = new ArrayList<>(members);
        //根据成员id集合查询各自对应的该群已读最大消息id
//        List<GChatReadPO> readList = gChatReadMapper.selectListByIds(ids, messageRequestBean.getGroupId());
        //将列表通过流组装成map<uid,maxid>
//        Map<String, GChatReadPO> collect = readList.parallelStream().collect(Collectors.toMap(GChatReadPO::getTo, Function.identity()));

        //组装消息
//        String jsonStr = UserUtils.getMessageResponseList(messageRequestBean, BaseConstants.GROUP_PREFIX);

        MessageResponseBean groupChat = this.getGroupChat(messageRequestBean.getMessageId(), messageRequestBean.getFrom(), messageRequestBean.getGroupId(), BaseConstants.GROUP_PREFIX + messageRequestBean.getGroupId(),
                messageRequestBean.getAvatarUrl(), messageRequestBean.getUserName(), messageRequestBean.getData(), messageRequestBean.getCreateTime(),
                baseGroupInfoPO.getName(), baseGroupInfoPO.getPic(), messageRequestBean.getCode(),
                messageRequestBean.getWidth(), messageRequestBean.getHeight(), messageRequestBean.getSoundSeconds(), messageRequestBean.getToMembers(),
                messageRequestBean.getPayChannel()
        );
//        MessageResponseBean groupChat = this.getGroupChat(messageRequestBean.getMessageId(), messageRequestBean.getFrom(), messageRequestBean.getGroupId(), BaseConstants.GROUP_PREFIX + messageRequestBean.getGroupId(),
//                messageRequestBean.getAvatarUrl(), messageRequestBean.getUserName(), messageRequestBean.getData(), messageRequestBean.getCreateTime(),
//                baseGroupInfoPO.getName() + "(" + members.size() + ")", baseGroupInfoPO.getPic(), messageRequestBean.getCode(),
//                messageRequestBean.getWidth(), messageRequestBean.getHeight(), messageRequestBean.getSoundSeconds(), messageRequestBean.getToMembers(),
//                messageRequestBean.getPayChannel()
//        );
        Map<String, Boolean> userSetMapinfo = null;
        try {
            userSetMapinfo = userSetMap.get();
//            terminalMapinfo = terminalMap.get();
        } catch (Exception e) {
            e.printStackTrace();
            SystemError.wrapBs(EnumType.ResponseCode.Fail.getCode(), "发送失败");
        }
        String jsonStr = JsonUtil.toJson(new ArrayList<>(Collections.singletonList(groupChat)));

        //便利ids，先推送消息，在判定是否存在已读数据，不存在则添加新的数据，存在则更新最大已读id
        Map<String, Boolean> finalUserSetMapinfo = userSetMapinfo;

        //添加需要 @ 的成员
//        if (StringHandler.isNotEmpty(messageRequestBean.getToMembers())) {
//            Arrays.stream(messageRequestBean.getToMembers().split(",")).parallel().map(x -> String.valueOf(McodeHandler.from32Radix(x))).forEach(list::add);
//        }

        Set<String> pushMembers = new HashSet<>();
        ids.forEach(obj -> {
            boolean flag = false;
            ServerWebSocket serverWebSocket = ChatObjectManager.getConnectionMap().get(McodeHandler.to32Radix(Long.parseLong(obj)));
            if (serverWebSocket != null && !obj.equals(fromId)) {
                try {
//                    if (currentTime - ChatObjectManager.getConnectionMapTime().get(McodeHandler.to32Radix(Long.parseLong(obj))) <= 1500) {
                    if (!ChatObjectManager.getConnectionUnReadPage().contains(obj)) {
                        flag = WebSocketVerticle.sendMessage(jsonStr, serverWebSocket);
                    }
//                    } else {
//                        serverWebSocket.close();
//                        ChatObjectManager.removeChatMapByKey(McodeHandler.to32Radix(Long.parseLong(obj)));
//                    }
                } catch (Exception e) {
                    Logger.info("发送群聊消息出现异常：消息编号【" + messageRequestBean.getCode() + "],当前群聊编号【" + messageRequestBean.getGroupId() + "】,接收人：" + obj);
                }
            }

            //对于本人来说  虽然不收到推送  但是这条消息也是读过的
            if (obj.equals(fromId)) {
                flag = true;
            }
            if (!obj.equals(fromId)) {
                String userUnReadMessageEntityKey = UserKey.userUnReadMessageEntityKey(obj);
                Redis.hash.hSet(userUnReadMessageEntityKey, gChatPO.getMessageId(), JsonUtil.toJson(groupChat));
            }
//            }

            if (!flag) {
                //判断用户是否开启免打扰
                if (finalUserSetMapinfo != null && finalUserSetMapinfo.get(obj) != null && !finalUserSetMapinfo.get(obj)) {
                    if (StringHandler.isNotEmpty(messageRequestBean.getToMembers())) {
                        if (messageRequestBean.getToMembers().contains(BaseConstants.PUSH_ALL)) {
                            //@所有人
//                        set.add(obj);
                            pushMembers.add(obj);
                        } else {
                            if ((messageRequestBean.getToMembers().contains(McodeHandler.to32Radix(StringHandler.getLong(obj))))) {
                                //@指定人
//                            set.add(obj);
                                pushMembers.add(obj);
                            }
                        }
                    }

                }
                //判断用户是否开启免打扰
//                if (finalUserSetMapinfo != null && finalUserSetMapinfo.get(obj) != null && !finalUserSetMapinfo.get(obj)) {
////                    set.add(obj);
//                    pushMembers.add(obj);
//                }
            }

        });

        ThreadUtil.execAsync(() -> {
            if (!pushMembers.isEmpty()) {
                JPushUtils.sendNotification(pushMembers, messageRequestBean);
            }

        });

    }

    @Override
    public void sendMessageSystem(MessageRequestBean messageRequestBean) {
        //处理系统消息：转发给所有用户
        List<ServerWebSocket> serverWebSockets = new ArrayList<>(ChatObjectManager.getConnectionMap().values());
        serverWebSockets.stream().parallel().forEach(o -> {
            if (o != null) {
//                WebSocketVerticle.sendMessage(o, messageRequestBean, BaseConstants.GROUP_PREFIX);
            }
        });
    }

    @Override
    public void clickToReadMessage(Long uid) {
        sChatMapper.batchUpdateIsRead(uid, uid % BaseConstants.sChatTableNum);
    }

    @Override
    public List<MessageResponseBean> getSingleChatUnRead(Long uid) {
        long suffix = uid % BaseConstants.sChatTableNum;
        SChatPO sChatPO = new SChatPO();

        sChatPO.setToId(uid);
        sChatPO.setTableName(BaseConstants.S_TABLE_PREFIX + suffix);
        sChatPO.setIsSuccess(false);
        //查询未发送成功的
        List<SChatPO> list = sChatMapper.select(sChatPO);
        List<MessageResponseBean> result = new ArrayList<>();
        if (!list.isEmpty()) {
            list.forEach(o -> {
                //组装消息对象
                String userId = McodeHandler.to32Radix(o.getFromId());
                MessageResponseBean obj = this.getSingleChat(o.getMessageId(), userId, BaseConstants.SINGLE_PREFIX + userId, o.getFromAvatarUrl(),
                        o.getFromName(), o.getData(), o.getCreateTime(), o.getCode(), o.getWidth(), o.getHeight(), o.getSoundSeconds(),"0",3);
                result.add(obj);
            });
        }
        return result;
    }

    @Override
    public List<MessageResponseBean> getUserApplyList(Long uid) {
        List<MessageResponseBean> result = new ArrayList<>();
        //查询 redis 中好友申请信息
        String key = UserKey.getUserApplyKey(uid);
        if (Redis.key.exists(key)) {
            Map map = Redis.hash.hgetall(key);
            if (map != null && !map.isEmpty()) {
                //转换成 list
                List<String> arrayList = new ArrayList<String>(map.values());
                if (StringHandler.isNotEmpty(arrayList)) {
                    List<BaseUserApplyPO> list = arrayList
                            .stream()
                            .map(s -> JSONObject.parseObject(s, BaseUserApplyPO.class))
                            .collect(Collectors.toList());
                    if (!list.isEmpty()) {
                        list.stream().filter(o -> !o.getIsDeal()).forEach(o -> {
                            //创建消息对象
                            LoginInfo cacheUser = UserUtils.getCacheUser(o.getFriendId());
                            String friendId = McodeHandler.to32Radix(Long.parseLong(o.getFriendId()));
                            MessageResponseBean obj = this.getSingleChat(IdUtil.simpleUUID(), friendId, BaseConstants.SINGLE_PREFIX + friendId,
                                    cacheUser.getAvatarUrl(), cacheUser.getNickName(), null, null, EnumType.MessageCode.MESSAGE_USER_APPLY.getCode(),
                                    0.0, 0.0, null,"0",3);
                            result.add(obj);
                        });
                    }
                }
            }
        } else {
            List<BaseUserApplyPO> list = userApplyMapper.getApplyList(String.valueOf(uid));
            if (StringHandler.isNotEmpty(list) && !list.isEmpty()) {
                //保存缓存
                Map<String, String> collect = list
                        .stream()
                        .collect(Collectors.toMap(BaseUserApplyPO::getFriendId, JsonUtil::toJson));
                Redis.hash.hmset(key, collect);
                Redis.key.setKeyTimeout(key, BaseConstants.FriendsConstants.SAVE_TIME_APPLY);
                list.stream().filter(o -> !o.getIsDeal()).forEach(o -> {
                    LoginInfo cacheUser = UserUtils.getCacheUser(o.getFriendId());
                    String friendId = McodeHandler.to32Radix(Long.parseLong(o.getFriendId()));
                    MessageResponseBean obj = this.getSingleChat(IdUtil.simpleUUID(), friendId, BaseConstants.SINGLE_PREFIX + friendId,
                            cacheUser.getAvatarUrl(), cacheUser.getNickName(), null, null, EnumType.MessageCode.MESSAGE_USER_APPLY.getCode(),
                            0.0, 0.0, null,"0",3);
                    result.add(obj);
                });
            }
        }
        return result;
    }

    @Override
    public List<MessageResponseBean> getGroupChatUnRead(Long uid) {
        List<GChatPO> result = new ArrayList<>();

        //判断用户在那些组
        //尝试在缓存中获取，缓存中没有在数据库中获取
        String key = UserKey.groupInfoKey(uid);
        List<String> groupIds = null;
        if (Redis.key.exists(key)) {
            Map cacheMap = Redis.hash.hgetall(key);
            groupIds = new ArrayList<String>(cacheMap.keySet());
        } else {
            //查询用户有哪些组
            List<BaseGroupInfoPO> userGroupListById = groupInfoMapper.findUserGroupListById(uid);
            if (userGroupListById != null && !userGroupListById.isEmpty()) {
                groupIds = userGroupListById.stream().map(o -> String.valueOf(o.getId())).collect(Collectors.toList());
                //现存一个用户相关群组的缓存    在用户获取群组详情的时候修改
                Map<String, String> collect = userGroupListById.stream().collect(Collectors.toMap(o -> String.valueOf(o.getId()), JsonUtil::toJson)/*,(v1,v2)-> v2)*/);
                Redis.hash.hmset(key, collect);
                Redis.key.setKeyTimeout(key, BaseConstants.GroupConstants.SAVE_TIME_USER_GROUP);
            }
        }
        Map<String, GChatPO> groupMaxId = new HashMap<>();
        if (groupIds != null && !groupIds.isEmpty()) {
            Map<Long, List<String>> collect = groupIds.stream().collect(Collectors.groupingBy(o -> Long.parseLong(o) % BaseConstants.gChatTableNum));
            collect.entrySet().forEach(x -> {
                x.getValue().forEach(groupId -> {
                    List<GChatPO> list = gChatMapper.findUnReadGroupMessageByGroupId(x.getKey(), groupId, uid);
                    if (StringHandler.isNotEmpty(list) && !list.isEmpty()) {
                        groupMaxId.put(groupId, list.get(list.size() - 1));
                    }
                    result.addAll(list);
                });
            });
        }

        if (!result.isEmpty()) {
            //修改最大id值  分组排序，取id最大
            //群id->最大id的消息
//            Map<Long, GChatPO> map = result.parallelStream().collect(
//                    Collectors.toMap(GChatPO::getTo, Function.identity(), (o1, o2) -> o1.getId() > o2.getId() ? o1 : o2)
//            );
            //批量更新最大已读id  放缓存
            List<GChatPO> gChatPOList = new ArrayList<>(groupMaxId.values());
            Redis.str.set(UserKey.groupChatUpdateCacheKey(uid), JsonUtil.toJson(gChatPOList), 5 * 60);

            //组装返回的数据
            List<Object> collect = result.stream().map(x -> String.valueOf(x.getTo())).distinct().collect(Collectors.toList());
            List<Object> groupStrList = Redis.hash.hmget(key, collect);
            Map<String, BaseGroupInfoPO> groupMap = null;
            if (StringHandler.isNotEmpty(groupStrList)) {
                groupMap = groupStrList.stream().map(x -> JsonUtil.jsonToBean((String) x, BaseGroupInfoPO.class))
                        .filter(x -> StringHandler.isNotEmpty(x.getId()))
                        .collect(Collectors.toMap(x -> String.valueOf(x.getId()), Function.identity()));
            } else {
                //查询用户有哪些组
                List<BaseGroupInfoPO> userGroupListById = groupInfoMapper.findUserGroupListById(uid);
                groupMap = userGroupListById.stream().collect(Collectors.toMap(x -> String.valueOf(x.getId()), Function.identity()));
            }

            Map<String, BaseGroupInfoPO> finalGroupMap = groupMap;
            return result.stream().map(o -> {
//                MessageResponseBean response = UserUtils.getMessageResponse(String.valueOf(o.getFrom()), String.valueOf(o.getTo()),
//                        o.getCode(), o.getData(), o.getCreateTime(), o.getWidth(), o.getHeight());
                BaseGroupInfoPO baseGroupInfoPO = finalGroupMap.get(String.valueOf(o.getTo()));
                String fromId = McodeHandler.to32Radix(o.getFrom());
                MessageResponseBean response = this.getGroupChat(o.getMessageId(), fromId, String.valueOf(o.getTo()), BaseConstants.GROUP_PREFIX + o.getTo(), o.getFromAvatarUrl(), o.getFromName(),
                        o.getData(), o.getCreateTime(), baseGroupInfoPO.getName(), baseGroupInfoPO.getPic(), o.getCode(), o.getWidth(), o.getHeight(), o.getSoundSeconds(), o.getToMembers(),3);
                return response;
            }).collect(Collectors.toList());
        } else {
            return null;
        }
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void updateChat(Long uid) {
        //更新单聊信息
        long suffix = uid % BaseConstants.sChatTableNum;
        sChatMapper.batchUpdateIsRead(uid, suffix);

        //更新群聊信息
        String jsonStr = Redis.str.get(UserKey.groupChatUpdateCacheKey(uid));
        Redis.key.del(UserKey.groupChatUpdateCacheKey(uid));
        if (StringHandler.isNotEmpty(jsonStr)) {
            JSONArray objects = JSONArray.parseArray(jsonStr);
            objects.forEach(o -> {
                GChatPO gChatPO = JSONObject.parseObject(JsonUtil.toJson(o), GChatPO.class);
                GChatReadPO obj = gChatReadMapper.selectObject(gChatPO.getTo(), uid);
                if (StringHandler.isEmpty(obj)) {
                    obj = new GChatReadPO();
                    obj.setFrom(String.valueOf(gChatPO.getTo()));
                    obj.setTo(String.valueOf(uid));
                    obj.setMaxId(gChatPO.getId());
                    gChatReadMapper.insertSelective(obj);
                } else {
                    obj.setMaxId(gChatPO.getId());
                    gChatReadMapper.updateByPrimaryKeySelective(obj);
                }
//                gChatReadMapper.updateGChatReadList(gChatPO.getId(), gChatPO.getTo(), uid);
            });
//            List<GChatPO> list = JsonUtil.jsonToBean(jsonStr, List.class);
//            list.parallelStream().forEach(o -> gChatReadMapper.updateGChatReadList(o.getId(), o.getTo(), uid));
        }
    }

    @Override
    public List<MessageResponseBean> getUserNoticeList(Long uid) {
        String key = UserKey.noticeKey(uid);
        List<String> range = Redis.list.range(key, 0L, -1L);
        if (range != null && !range.isEmpty()) {
            //删除通知key
            Redis.key.del(key);
            return range.stream().map(o -> JsonUtil.jsonToBean(o, MessageResponseBean.class)).collect(Collectors.toList());
        }
        return null;
    }

    @Override
    public List<MessageResponseBean> getClearList(Long uid) {
        String key = UserKey.messageClearKey(uid);
        Map result = Redis.hash.hgetall(key);
        if (result != null && !result.isEmpty()) {
            Redis.key.del(key);
            List<String> list = new ArrayList<String>(result.values());
            return list.stream().map(o -> JsonUtil.jsonToBean(o, MessageResponseBean.class)).collect(Collectors.toList());
        }
        return null;
    }

    /**
     * @ 名称:组装消息对象
     * @ 创建人: dhf
     * @ 参数： messageId
     * @ 参数： fromId
     * @ 参数： windowId
     * @ 参数： avatarUrl
     * @ 参数： userName
     * @ 参数： data
     * @ 参数： createTime
     * @ 参数： code
     * @ 参数： width
     * @ 参数： height
     * @ 参数： soundSeconds
     * @ 创建时间: 2022/5/11
     * @ 修改人和其它信息
     * @ 返回值: com.ys.web.ws.bean.MessageResponseBean
     */
    public MessageResponseBean getSingleChat(String messageId, String fromId, String windowId, String avatarUrl, String userName, String data, String createTime,
                                             Integer code, Double width, Double height, String soundSeconds,String level,int payChannel) {

        MessageResponseBean result = new MessageResponseBean();
        result.setMessageId(messageId);
        result.setFrom(fromId);
        result.setWindowId(windowId);
        result.setCreateTime(createTime);
        result.setAvatarUrl(avatarUrl);
        result.setUserName(userName);
        result.setData(data);
        result.setCode(code);
        result.setHeight(height == null ? 0.0 : height);
        result.setWidth(width == null ? 0.0 : width);
        result.setSoundSeconds(soundSeconds);
        result.setLevel(StringHandler.isEmpty(level)?"0":level);
        result.setPayChannel(payChannel);
        return result;
    }

    /**
     * @ 名称:组装消息对象
     * @ 创建人: dhf
     * @ 参数： messageId
     * @ 参数： fromId
     * @ 参数： groupId
     * @ 参数： windowId
     * @ 参数： avatarUrl
     * @ 参数： userName
     * @ 参数： data
     * @ 参数： createTime
     * @ 参数： groupName
     * @ 参数： groupPic
     * @ 参数： code
     * @ 参数： width
     * @ 参数： height
     * @ 参数： soundSeconds
     * @ 参数： toMembers
     * @ 创建时间: 2022/5/11
     * @ 修改人和其它信息
     * @ 返回值: com.ys.web.ws.bean.MessageResponseBean
     */
    public MessageResponseBean getGroupChat(String messageId, String fromId, String groupId, String windowId, String avatarUrl, String userName, String data,
                                            String createTime, String groupName, String groupPic, Integer code, Double width, Double height, String soundSeconds,
                                            String toMembers,int payChannel) {
        MessageResponseBean result = new MessageResponseBean();
        result.setMessageId(messageId);
        result.setFrom(fromId);
        result.setGroupId(groupId);
        result.setWindowId(windowId);
        result.setAvatarUrl(avatarUrl);
        result.setUserName(userName);
        result.setData(data);
        result.setCreateTime(createTime);
        result.setGroupName(groupName);
        result.setCode(code);
        result.setHeight(height == null ? 0.0 : height);
        result.setWidth(width == null ? 0.0 : width);
        result.setGroupPic(groupPic);
        result.setSoundSeconds(soundSeconds);
        result.setToMembers(toMembers);
        result.setPayChannel(payChannel);
        return result;
    }

    @Override
    public void undoMessage(String uid, MessageRequestBean messageRequestBean) {
        if (messageRequestBean.getFrom().equals(UserUtils.ASSISTANT.getUserId())) {
            SystemError.wrapBs("500", "撤销失败！");
        }
        MessageResponseBean messageResponseBean = new MessageResponseBean();
        messageResponseBean.setMessageId(messageRequestBean.getMessageId());
        messageResponseBean.setCode(EnumType.MessageCode.MESSAGE_GROUP_UNDO_MESSAGE.getCode());
        messageResponseBean.setFrom(messageRequestBean.getFrom());
        messageResponseBean.setUserName(messageRequestBean.getUserName());
        messageResponseBean.setCreateTime(DateHandler.getDate(new Date(), BaseConstants.BASE_FORMAT));
        messageResponseBean.setAvatarUrl(messageRequestBean.getAvatarUrl());

        String groupId = messageRequestBean.getGroupId();
        List<String> members;
        if (StringHandler.isNotEmpty(groupId)) {
            members = new ArrayList<>(Redis.zSet.range(GroupKey.groupMembersKey(groupId), 0, -1));
            messageResponseBean.setGroupId(groupId);
            BaseGroupInfoPO baseGroupInfoPO = GroupUtils.getCacheGroupInfo(groupId);
            messageResponseBean.setGroupPic(baseGroupInfoPO.getPic());
//            messageResponseBean.setGroupName(baseGroupInfoPO.getName() + "(" + members.size() + ")");
            messageResponseBean.setGroupName(baseGroupInfoPO.getName());
            messageResponseBean.setWindowId(BaseConstants.GROUP_PREFIX + groupId);
            String fromId = String.valueOf(McodeHandler.from32Radix(messageRequestBean.getFrom()));
            if (uid.equals(fromId)) {
                //id 相同  撤回消息的人和发送消息的人是同一个人
                messageResponseBean.setData(TipUtils.undoMessageTip(messageRequestBean.getUserName()));
            } else {
                //id 不相同  撤回消息的人和发送消息的人不是同一个人
                Double uScore = Redis.zSet.score(GroupKey.groupMembersKey(groupId), uid);
                Double fScore = Redis.zSet.score(GroupKey.groupMembersKey(groupId), fromId);
                if (StringHandler.isEmpty(fScore)) {
                    fScore = 3.0;
                }
                String prefix = uScore.compareTo(EnumType.GroupIdentity.MASTER.getCode().doubleValue()) == 0 ? "\"群主\"" : "\"管理员\"";
                if (uScore <= fScore) {
                    messageResponseBean.setData(prefix + "撤回了一条消息");
                } else {
                    SystemError.wrapBs("500", "无法撤回群主的信息");
                }
                String from = McodeHandler.to32Radix(StringHandler.getLong(uid));
                messageRequestBean.setFrom(from);
                messageResponseBean.setFrom(from);
            }
            members = members.stream().filter(x -> !x.equals(String.valueOf(McodeHandler.from32Radix(messageRequestBean.getFrom())))).collect(Collectors.toList());
        } else {
            members = new ArrayList<>();
            members.add(String.valueOf(McodeHandler.from32Radix(messageRequestBean.getTo())));
            messageResponseBean.setWindowId(BaseConstants.SINGLE_PREFIX + messageRequestBean.getFrom());
            messageResponseBean.setData(TipUtils.undoMessageTip("\"" + messageRequestBean.getUserName() + "\""));
        }

        members.forEach(o -> GroupUtils.sendIfAbsent(messageResponseBean, o, messageResponseBean.getWindowId()));
//        GroupUtils.sendIds(members,messageResponseBean);

        List<String> finalMembers = members;
        ThreadUtil.execAsync(() -> {
            //撤回对应消息
            finalMembers.forEach(x -> Redis.hash.delHash(UserKey.userUnReadMessageEntityKey(x), messageRequestBean.getMessageId()));

        });


    }


    @Override
    public List<MessageResponseBean> queryHistoryChatGroup(Long uid, QueryHistoryVo queryHistoryVo) {
        long suffix = StringHandler.getLong(queryHistoryVo.getQueryId()) % BaseConstants.gChatTableNum;
        if (StringHandler.isNotEmpty(queryHistoryVo.getMembersId())) {
            queryHistoryVo.setMembersId(StringHandler.getString(McodeHandler.from32Radix(queryHistoryVo.getMembersId())));
        }
        List<GChatPO> gChatPOList = gChatMapper.queryHistoryChatGroup(uid, queryHistoryVo, suffix);
        if (gChatPOList.isEmpty()) {
            return new ArrayList<>();
        }
        return gChatPOList.stream().map(o -> {
            String fromId = McodeHandler.to32Radix(o.getFrom());
            return this.getGroupChat(o.getMessageId(), fromId, String.valueOf(o.getTo()), BaseConstants.GROUP_PREFIX + o.getTo(), o.getFromAvatarUrl(), o.getFromName(),
                    o.getData(), o.getCreateTime(), null, null, o.getCode(), o.getWidth(), o.getHeight(), o.getSoundSeconds(), o.getToMembers(),3);
        }).collect(Collectors.toList());
    }

    @Override
    public List<MessageResponseBean> queryHistorySingleChat(Long uid, QueryHistoryVo queryHistoryVo) {
        long userSuffix = uid % BaseConstants.sChatTableNum;
        long friendSuffix = StringHandler.getLong(queryHistoryVo.getQueryId()) % BaseConstants.sChatTableNum;
        List<SChatPO> sChatPOList = sChatMapper.queryHistorySingleChat(uid, queryHistoryVo, userSuffix, friendSuffix);
        if (sChatPOList.isEmpty()) {
            return new ArrayList<>();
        }
        String userId = McodeHandler.to32Radix(StringHandler.getLong(queryHistoryVo.getQueryId()));
        return sChatPOList.stream().map(o -> {
            //组装消息对象
            return this.getSingleChat(o.getMessageId(), userId, BaseConstants.SINGLE_PREFIX + userId, o.getFromAvatarUrl(),
                    o.getFromName(), o.getData(), o.getCreateTime(), o.getCode(), o.getWidth(), o.getHeight(), o.getSoundSeconds(),"0",3)
                    ;
        }).collect(Collectors.toList());
    }


    @Override
    public void freezeGroupAndUserCreate() {
        List<Map<String, String>> list = userGroupMapper.findData();
        System.err.println(JsonUtil.toJson(list));

        List<String> userIdList = list.stream().map(x -> x.get("user_id")).collect(Collectors.toList());
//        删除用户创建群聊申请
        userGroupMapper.deleteGroupApply(userIdList);

        //禁用群聊
        list.forEach(x -> {
            int row = userGroupMapper.updateInfoById(x.get("id"), 1);
            if (row == 1) {
                String key = GroupKey.groupInfoKey(x.get("id"));
                BaseGroupInfoPO baseGroupInfoPO = JsonUtil.jsonToBean(Redis.str.get(key), BaseGroupInfoPO.class);
                baseGroupInfoPO.setIsFreeze(true);
                Redis.str.set(key, JsonUtil.toJson(baseGroupInfoPO));
            }
        });

    }


    @Override
    public void query() {
        List<BaseGroupInfoPO> groupInfoPOList = groupInfoMapper.selectAll();
        groupInfoPOList.forEach(x -> {
            Set<String> range = Redis.zSet.range(GroupKey.groupMembersKey(x.getId()), 0, -1);
            if (range.contains("13050")) {
                System.err.println(JsonUtil.toJson(x));
            }
        });
    }

    @Override
    public void groupMessage() {
        Set<String> range = Redis.zSet.range(GroupKey.groupMembersKey(335), 0, -1);
        range.forEach(x -> {
            LoginInfo cacheUser = UserUtils.getCacheUser(x);
            System.out.println("-------" + cacheUser.getNickName() + "--------");
            String userUnReadMessageEntityKey = UserKey.userUnReadMessageEntityKey(x);
            if (Redis.key.exists(userUnReadMessageEntityKey)) {
                Map hgetall = Redis.hash.hgetall(userUnReadMessageEntityKey);
                if (!hgetall.isEmpty()) {
                    List<String> groupChatList = new ArrayList<String>(hgetall.values());
                    groupChatList.forEach(o -> {
                        try {
                            JsonUtil.jsonToBean(o, MessageResponseBean.class);
                            System.out.println(o);
                        } catch (Exception e) {
                            System.err.println(o);
                        }
                    });
                }
            }

            System.out.println("-------------------------------------");
        });
    }
}
